package edu.hit.crawler.io;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

@SuppressWarnings("deprecation")
public class DbReader extends Configured implements Tool {

	public static class DbReaderReducer extends
			Reducer<Text, CrawItem, Text, Text> {

		@Override
		public void reduce(Text key, Iterable<CrawItem> values, Context context)
				throws IOException, InterruptedException {

			for (CrawItem it : values) {
				if (it.getStatus() == CrawItem.SUCCESS) {
					System.out.println(it.getLastModified() + " " + it.getLastFetchTime());
					context.write(key, new Text(it.toString()));
				}
			}
		}

	}

	/**
	 * @param args
	 * @throws Exception
	 */
	public static void main(String[] args) throws Exception {
		Configuration conf = new Configuration();
		System.exit(ToolRunner.run(conf, new DbReader(), args));
	}

	@Override
	public int run(String[] args) throws Exception {

		Configuration conf = getConf();
		Job job = new Job(conf, "DbRead");

		job.setReducerClass(DbReaderReducer.class);

		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(CrawItem.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(Text.class);

		job.setInputFormatClass(SequenceFileInputFormat.class);
		job.setOutputFormatClass(TextOutputFormat.class);
		
		job.setNumReduceTasks(2);

		FileSystem fs = FileSystem.get(conf);
		String workdir = conf.get("org.work.crawler.dir", "crawler/");

		long latest = 0;
		Path in = null;
		// the crawl_db
		for (FileStatus stat : fs.listStatus(new Path(workdir + "crawl_db/"))) {
			if (stat.getModificationTime() > latest) {
				in = stat.getPath();
				latest = stat.getModificationTime();
			}
		}
		SequenceFileInputFormat.addInputPath(job, in);
		TextOutputFormat.setOutputPath(job, new Path(workdir + "db_text"));

		return job.waitForCompletion(true) ? 0 : 1;

	}

}
